package com.wanke.publish;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.wanke.base.dto.ResultObject;
import com.wanke.config.HttpConfig;
import com.wanke.constant.ProtocolConstants;
import com.wanke.domain.po.Measurement;
import com.wanke.domain.po.Measurements;
import io.vertx.core.Future;
import io.vertx.core.json.Json;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.wanke.service.MeasurementService;

import java.time.Instant;
import java.time.LocalDateTime;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicLong;

import static com.wanke.config.HttpConfig.authClient;
import static com.wanke.config.HttpConfig.measureClient;

/**
 * http 量测上传程序
 * @author diaoshili
 * @date 2021-02-02
 */
public class HttpRemoteMeasurePublish extends RemoteMeasurePublish {

    public static AtomicLong systemHeartBeatTime = new AtomicLong(0);

    private static final ProtocolConstants protocol = ProtocolConstants.HTTP;

    private final MeasurementService measurementService;

    private int pollSize = 500;

    private final Long tokenExprie = 600L;

    private Long lastTokenTime = -1L;

    private final String username;

    private final String password;

    private String token;

    private final Long maxTimeOut;

    private final Long stationCode;

    private final String type;

    private final Long maxTimeOutPoint;

    private final Long frequency;

    private final ObjectMapper objectMapper = new ObjectMapper();

    private final String httpHost;

    private static final Logger logger = LoggerFactory.getLogger(HttpRemoteMeasurePublish.class);

    public HttpRemoteMeasurePublish(MeasurementService measurementService,
                                    String username, String password, String type,
                                    Long maxTimeOut, Long maxTimeOutPoint, Long stationCode, Long frequency, int pollSize, String httpHost) {
        this.measurementService = measurementService;
        this.username = username;
        this.type = type;
        this.password = password;
        this.maxTimeOut = maxTimeOut * 1000;
        this.maxTimeOutPoint = maxTimeOutPoint;
        this.frequency = frequency * 1000;
        this.stationCode = stationCode;
        this.httpHost = httpHost;
        this.pollSize = pollSize;
    }

    @Override
    public void onTempStore(Measurements measurements) {
        measurementService.insertTempMeasurement(measurements.getMeasurementList());
    }

    @Override
    public boolean isConnect() {
        return (Instant.now().toEpochMilli() - systemHeartBeatTime.get()) < maxTimeOut;
    }

    @Override
    public void heartBeatStart() {
        vertx.setPeriodic(frequency, time -> {
            HttpConfig.init(vertx, httpHost);
            long epochSecond = Instant.now().getEpochSecond();
            if ((epochSecond - lastTokenTime) > tokenExprie * 1000) {
                Future<ResultObject<Map<String, Object>>> login = null;
                try {
                    login = authClient.login(username, password, type);
                } catch (Exception e) {
                    e.printStackTrace();
                }
                Future<ResultObject<Map<String, Object>>> finalLogin = login;
                login.onComplete(a -> {
                    if (a.succeeded()) {
                        try {
                            ResultObject<Map<String, Object>> userResultObject = finalLogin.result();
                            if (userResultObject.getResults() != null && userResultObject.getResults().get("token") != null) {
                                token = userResultObject.getResults().get("token").toString();
                                lastTokenTime = epochSecond;
                            }
                        } catch (Exception e) {
                            logger.error("登陆时发生错误", e);
                        }
                    } else {
                        logger.error("登陆时发生错误", finalLogin.cause());
                    }
                });
            }
            measureClient.updateHeartbeat(token, maxTimeOutPoint).onComplete(a -> {
                if (a.succeeded()) {
                    logger.info("更新心跳时成功");
                } else {
                    logger.error("更新心跳时发生错误", a.cause());
                }
            });
        });
    }

    @Override
    public void toSend(Measurements measurements) {
        List<Measurement> measure = measurements.getMeasurementList();
        //发布云端消息
        Future<Map<String, Object>> result = measureClient.loadMeasure(token, measure);
        result.onComplete(a -> {
            if (result.succeeded()) {
                logger.debug("量测消息上传成功,上传量测数量{}", measurements.getMeasurementList().size());
            } else {
                logger.error("保存时发生错误", result.cause());
                measurementService.insertTempMeasurement(measure);
            }
        });
    }

    @Override
    public void resendStart() {
        vertx.setPeriodic(frequency * 12, time -> {
            logger.info("开始检查HTTP断线数据", time);
            if (this.isConnect()) {
                LocalDateTime localDateTime = LocalDateTime.now();
                this.resent(localDateTime);
            }
        });
    }

    public void resent(LocalDateTime time) {
        try {
            logger.info("开始检查HTTP断线数据", time);
            Long total = measurementService.findMeasurementTempAllSize(time);
            logger.info("开始被动补发http消息量测,代发数量{}", total);
            if (total != null && total > 0) {
                logger.info("历史曾经断线,重发过去的数据：{},信息{}", time);
                long v = total / pollSize + 1;
                for (int i = 0; i <= v ; i++) {
                    if (!isConnect()) {
                        logger.info("HTTP网络断线无法进行重发");
                        return;
                    }
                    List<Measurement> all = measurementService.findTempAll(time, pollSize, i * pollSize);
                    if (all != null || !all.isEmpty()) {
                        int finalI = i;
                        all.forEach(a -> {
                            logger.warn("重传量测数据第{}次,传输条数{},总条数{}", finalI, all.size(), all.size());
                            measureClient.recollect(token, all);
                        });
                    }
                }
                measurementService.deleteTempMeausremnet(time);
            }
        } catch (Exception e) {
            logger.error("点号数据重传消息处理失败 错误原因", e);
        }
    }

    @Override
    public ProtocolConstants getProtocol() {
        return protocol;
    }

    @Override
    public MeasurementService getMeasurementService() {
        return measurementService;
    }
}
